package com.jcode.netty.server;

import com.jcode.netty.core.AbstractRemotingService;
import com.jcode.netty.core.ChannelEventListener;
import com.jcode.netty.process.DefaultRequestProcessor;
import com.jcode.netty.codec.NettyDecoder;
import com.jcode.netty.codec.NettyEncoder;
import com.jcode.netty.serialize.Serializer;
import com.jcode.netty.serialize.SerializerFactory;
import com.jcode.netty.util.NamedThreadFactory;
import com.jcode.netty.util.RemotingUtil;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.AttributeKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @Desc
 * @Author dreajay
 * @DateTime 2016年10月15日 上午11:15:40
 */
public class NettyServer extends AbstractRemotingService {

    protected ServerBootstrap bootstrap;

    protected EventLoopGroup bossEventLoopGroup;

    protected EventLoopGroup selectorWorkerEventLoopGroup;

    // private DefaultEventExecutorGroup defaultBusinessEventExecutorGroup;

    protected AtomicBoolean start = new AtomicBoolean(false);

    protected ServerConfig config;

    protected ConnectionManager connectionManager;

    private static final Logger logger = LoggerFactory.getLogger(NettyServer.class);

    public NettyServer(ServerConfig config) {
        this(config, null);
    }

    public NettyServer(ServerConfig config, ChannelEventListener channelEventListener) {
        super(config.getServerOnewaySemaphoreValue(), config.getServerAsyncSemaphoreValue(), channelEventListener);
        this.config = config;
        ExecutorService defaultRequestExecutor = Executors.newFixedThreadPool(config.getServerWorkerThreads(), new NamedThreadFactory("defaultRequestExecutor"));
        registerDefaultProcessor(new DefaultRequestProcessor(), defaultRequestExecutor);
        setCallbackExecutor(defaultRequestExecutor);
        connectionManager = new ConnectionManager(this);

        // 已经在worker线程处理，这个线程池不需要，业务已经设置了自己的线程池
        // this.defaultBusinessEventExecutorGroup = new
        // DefaultEventExecutorGroup(config.getServerWorkerThreads(), new
        // NamedThreadFactory("defaultBusinessEventExecutorGroup"));

    }

    @Override
    public void start() {
        if (start.compareAndSet(false, true)) {
            this.bossEventLoopGroup = new NioEventLoopGroup(config.getServerBossThreads(), new NamedThreadFactory("bossEventLoopGroup"));

            if (RemotingUtil.isLinuxPlatform()
                    && config.isUseEpollNativeSelector()) {
                this.selectorWorkerEventLoopGroup = new EpollEventLoopGroup(config.getServerWorkerThreads(), new NamedThreadFactory("selectorWorkerEventLoopGroup"));
            } else {
                this.selectorWorkerEventLoopGroup = new NioEventLoopGroup(config.getServerWorkerThreads(), new NamedThreadFactory("selectorWorkerEventLoopGroup"));
            }

            final Serializer serializer = SerializerFactory.getSerializer(config.getSerializeType());
            bootstrap = new ServerBootstrap();
            bootstrap.group(bossEventLoopGroup, selectorWorkerEventLoopGroup)//
                    .channel(NioServerSocketChannel.class)//
                    .option(ChannelOption.SO_BACKLOG, 65536)//
                    .option(ChannelOption.SO_REUSEADDR, true)//
                    .option(ChannelOption.SO_KEEPALIVE, false)// TCP活跃检查
                    .childOption(ChannelOption.TCP_NODELAY, true)//
                    .option(ChannelOption.SO_SNDBUF, config.getServerSocketSndBufSize())//
                    .option(ChannelOption.SO_RCVBUF, config.getServerSocketRcvBufSize())//
                    .attr(AttributeKey.newInstance("net.protocol"), "customAttribute") // 添加扩展属性
                            // .childAttr(AttributeKey.newInstance("net.protocol"),
                            // Protocol.TCP)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel ch) throws IOException {
                            // 多协议处理
                            /*
							 * switch (config.getProtocol()) {
							 * case TCP:
							 * 
							 * break;
							 * case HTTP:
							 * 
							 * break;
							 * case WEBSOCKET:
							 * 
							 * break;
							 * default:
							 * break;
							 * }
							 */

                            // ch.pipeline().addLast(new
                            // LoggingHandler(LogLevel.INFO));
                            ch.pipeline().addLast(
                                    // defaultBusinessEventExecutorGroup,//
                                    new NettyEncoder(serializer),//
                                    new NettyDecoder(),//
                                    new IdleStateHandler(0, 0, config.getServerChannelMaxIdleTimeSeconds()),//
                                    new ServerHandler(NettyServer.this));
                        }
                    });

            bindAddress();

            super.start();
        }

    }

    @Override
    public void shutdown() {
        if (start.compareAndSet(true, false)) {
            super.shutdown();
            bossEventLoopGroup.shutdownGracefully();
            selectorWorkerEventLoopGroup.shutdownGracefully();
        }
    }

    protected void bindAddress() {
        bootstrap.bind(config.getListenPort()).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture f) throws Exception {
                if (f.isSuccess()) {
                    logger.error("Started Server success with listen port: {}", config.getListenPort());
                } else {
                    // 绑定监听端口，如果没有启动成功，尝试重启
                    logger.error("Started Server Failed, the port {} is already inuse, begin to restart", config.getListenPort());
					/*f.channel().eventLoop().schedule(new Runnable() {
						@Override
						public void run() {
							bindAddress();
						}
					}, 5, TimeUnit.SECONDS);*/
                }
            }
        });

    }

    public ConnectionManager getConnectionManager() {
        return connectionManager;
    }

    public void setConnectionManager(ConnectionManager connectionManager) {
        this.connectionManager = connectionManager;
    }

    public int localListenPort() {
        return config.getListenPort();
    }

    public ServerConfig getConfig() {
        return config;
    }

    public void setConfig(ServerConfig config) {
        this.config = config;
    }


}
